package stream

import (
	"bufio"
	"fmt"
	"io"
	"log"
	"os"
	"sync"
	"time"
)

var channelCacheSize = 1

func SetStreamChannelCacheSize(size int) {
	channelCacheSize = size
}

type (
	// Stream 定义数据流的接口
	Stream[T any] interface {
		// Map 处理（异步）
		Map(mapFunc func(item Item[T]) T) Stream[T]
		// Group 分组（异步）
		Group(groupFunc func(item Item[T]) string) Stream[T]
		// Distinct 去重（异步）
		Distinct(distinctFunc func(item Item[T]) any) Stream[T]
		// Filter 过滤（异步）
		Filter(filterFunc func(item Item[T]) bool) Stream[T]
		// Concat 聚合（异步）
		Concat(streams ...Stream[T]) Stream[T]
		// Count 统计（异步）
		Count(interval time.Duration, countChan chan<- *Count) Stream[T]
		// Head 收前 num 个元素（异步）
		Head(num int) Stream[T]
		// Tail 收后 num 个元素（异步）
		Tail(num int) Stream[T]
		// SampleByStepSize 根据步长采样（异步）
		SampleByStepSize(size int64) Stream[T]
		// SampleByTime 根据时间采样（异步）
		SampleByTime(interval time.Duration) Stream[T]
		// // Clone 克隆当前数据流（异步）
		// Clone() Stream[T]

		// Consume 一次性消费（异步）
		Consume(consumeFunc func(item Item[T]))
		// Drain 丢弃数据（同步）
		Drain(doBeforeFuncs ...func(item Item[T]))
		// Receive 读取元素（同步）
		Receive() (Item[T], bool)
		// Channel 获取数据流管道
		Channel() <-chan Item[T]
	}

	// item 定义数据流的数据结构
	stream[T any] struct {
		index  uint64
		source <-chan Item[T]
	}
)

// NewStreamOf 通过指定数据创建一个新的数据流
func NewStreamOf[T any](items ...Item[T]) Stream[T] {
	if len(items) == 0 {
		return nil
	}

	source := make(chan Item[T], channelCacheSize)
	stream := &stream[T]{index: 1, source: source}
	go func() {
		for _, item := range items {
			if item == nil {
				continue
			}

			source <- item.WithIndex(stream.index)
			stream.index++
		}

		close(source)
	}()

	return stream
}

// NewStreamBy 通过指定通道创建一个新的数据流
func NewStreamBy[T any](source <-chan Item[T]) Stream[T] {
	if source == nil {
		return nil
	}

	middle := make(chan Item[T], channelCacheSize)
	stream := &stream[T]{index: 1, source: middle}
	go func() {
		for item := range source {
			if item == nil {
				continue
			}

			middle <- item.WithIndex(stream.index)
			stream.index++
		}

		close(middle)
	}()

	return stream
}

// NewStreamWith 通过聚合多个流创建一个新的数据流
func NewStreamWith[T any](streams ...Stream[T]) Stream[T] {
	if len(streams) == 0 {
		return nil
	}

	source := make(chan Item[T], channelCacheSize)
	middle := streams[0].Concat(streams[1:]...)
	stream := &stream[T]{index: 1, source: source}
	go func() {
		for item := range middle.Channel() {
			if item == nil {
				continue
			}

			source <- item.WithIndex(stream.index)
			stream.index++
		}

		close(source)
	}()

	return stream
}

// NewStreamFrom 通过生成式函数创建一个新的数据流
func NewStreamFrom[T any](generator func() Stream[T]) Stream[T] {
	if generator == nil {
		return nil
	}

	source := make(chan Item[T], channelCacheSize)
	middle := generator()
	stream := &stream[T]{index: 1, source: source}
	go func() {
		for item := range middle.Channel() {
			if item == nil {
				continue
			}

			source <- item.WithIndex(stream.index)
			stream.index++
		}

		close(source)
	}()

	return stream
}

// NewStreamFromMap 将指定数据流处理后，输出到一个新的数据流
func NewStreamFromMap[T, R any](mapFunc func(item Item[T]) Item[R], streams ...Stream[T]) Stream[R] {
	if len(streams) == 0 {
		return nil
	}

	source := make(chan Item[R], channelCacheSize)
	middle := streams[0].Concat(streams[1:]...)
	stream := &stream[R]{index: 1, source: source}
	go func() {
		for item := range middle.Channel() {
			if item == nil {
				continue
			}

			item := mapFunc(item)
			if item == nil {
				continue
			}

			source <- item.WithIndex(stream.index)
			stream.index++
		}

		close(source)
	}()

	return stream
}

// NewStreamFromFile 将指定文件处理后，生成一个新的数据流
func NewStreamFromFile[T any](dealFunc func(line string) Item[T], filename string) (Stream[T], error) {
	if dealFunc == nil {
		return nil, fmt.Errorf("stream: nil deal function")
	} else if filename == "" {
		return nil, fmt.Errorf("stream: nil empty filename")
	}

	// 打开文件
	file, err := os.Open(filename)
	if err != nil {
		return nil, fmt.Errorf("stream: %v", err)
	}

	writer := func(deal func(line string) Item[T], line []byte, source chan<- Item[T], stream *stream[T]) {
		if item := deal(string(line)); item != nil {
			source <- item.WithIndex(stream.index)
			stream.index++
		}
	}

	lineNo := 0
	reader := bufio.NewReader(file)
	source := make(chan Item[T], channelCacheSize)
	stream := &stream[T]{source: source}
	go func() {
		for {
			lineNo++
			line, _, err := reader.ReadLine()
			if err != nil {
				// 文件读取完毕
				if err == io.EOF {
					if len(line) > 0 {
						writer(dealFunc, line, source, stream)
					}

					break
				}

				// 读取异常
				log.Printf("Failed to read the %d line: %v", lineNo, err)
			}

			if len(line) > 0 {
				writer(dealFunc, line, source, stream)
			}
		}

		close(source)
	}()

	return stream, nil
}

// Map 处理（异步）
func (own *stream[T]) Map(mapFunc func(item Item[T]) T) Stream[T] {
	if mapFunc == nil {
		return own
	}

	source := make(chan Item[T], channelCacheSize)
	go func() {
		for item := range own.source {
			source <- item.WithValue(mapFunc(item))
		}

		close(source)
	}()

	return &stream[T]{source: source}
}

// Group 分组（异步）
func (own *stream[T]) Group(groupFunc func(item Item[T]) string) Stream[T] {
	if groupFunc == nil {
		return own
	}

	source := make(chan Item[T], channelCacheSize)
	go func() {
		for item := range own.source {
			source <- item.WithGroup(groupFunc(item))
		}

		close(source)
	}()

	return &stream[T]{source: source}
}

// Distinct 去重（异步）
func (own *stream[T]) Distinct(distinctFunc func(item Item[T]) any) Stream[T] {
	if distinctFunc == nil {
		return own
	}

	source := make(chan Item[T], channelCacheSize)
	go func() {
		keys := make(map[any]struct{})
		for item := range own.source {
			key := distinctFunc(item)
			if _, ok := keys[key]; !ok {
				keys[key] = struct{}{}
				source <- item
			}
		}

		close(source)
	}()

	return &stream[T]{source: source}
}

// Filter 过滤（异步）
func (own *stream[T]) Filter(filterFunc func(item Item[T]) bool) Stream[T] {
	if filterFunc == nil {
		return own
	}

	source := make(chan Item[T], channelCacheSize)
	go func() {
		for item := range own.source {
			if !filterFunc(item) {
				source <- item
			}
		}

		close(source)
	}()

	return &stream[T]{source: source}
}

// Concat 聚合（异步）
func (own *stream[T]) Concat(streams ...Stream[T]) Stream[T] {
	if len(streams) == 0 {
		return own
	}

	source := make(chan Item[T], channelCacheSize)
	go func() {
		var wg sync.WaitGroup
		for _, stream := range append(streams, own) {
			if stream == nil {
				continue
			}

			wg.Add(1)
			channel := stream.Channel()
			go func(wg *sync.WaitGroup) {
				for item := range channel {
					source <- item
				}
				wg.Done()
			}(&wg)
		}

		wg.Wait()
		close(source)
	}()

	return &stream[T]{source: source}
}

type Count struct {
	Count int64     `json:"count" xml:"count"`
	Start time.Time `json:"start" xml:"start"`
	End   time.Time `json:"end" xml:"end"`
}

// Count 统计（异步）
func (own *stream[T]) Count(interval time.Duration, countChan chan<- *Count) Stream[T] {
	if interval < 1 || countChan == nil {
		return own
	}

	closed := false
	source := make(chan Item[T], channelCacheSize)
	go func() {
		count := &Count{Start: time.Now()}
		ticker := time.NewTicker(interval)
		for {
			select {
			case item, ok := <-own.source:
				if item != nil {
					count.Count++
				}

				if !ok {
					count.End = time.Now()
					countChan <- count
					closed = true
				}

				source <- item
			case <-ticker.C:
				count.End = time.Now()
				countChan <- count
				ticker.Reset(interval)
				count = &Count{Start: time.Now()}
			}

			if closed {
				break
			}
		}

		close(source)
		close(countChan)
	}()

	return &stream[T]{source: source}
}

// Head 只接收数据流前 num 个元素（异步）
func (own *stream[T]) Head(num int) Stream[T] {
	if num < 1 {
		return own
	}

	source := make(chan Item[T], channelCacheSize)
	go func() {
		var recv int
		for item := range own.source {
			source <- item
			if recv++; recv == num {
				break
			}
		}

		close(source)
	}()

	return &stream[T]{source: source}
}

// Tail 只接收数据流后 num 个元素（异步）
func (own *stream[T]) Tail(num int) Stream[T] {
	if num < 1 {
		return own
	}

	source := make(chan Item[T], channelCacheSize)
	go func() {
		index, items := 0, make([]Item[T], num)
		for item := range own.source {
			items[index%num] = item
			index++
		}

		for _, item := range items[index%num:] {
			source <- item
		}
		for _, item := range items[:index%num] {
			source <- item
		}

		close(source)
	}()

	return &stream[T]{source: source}
}

// SampleByStepSize 根据间隔采样（异步）
func (own *stream[T]) SampleByStepSize(size int64) Stream[T] {
	if size < 1 {
		return own
	}

	source := make(chan Item[T], channelCacheSize)
	go func() {
		interval := size
		for item := range own.source {
			if interval == size {
				source <- item
				interval = 0
				continue
			}

			interval++
		}

		close(source)
	}()

	return &stream[T]{source: source}
}

// SampleByTime 根据时间采样（异步）
func (own *stream[T]) SampleByTime(interval time.Duration) Stream[T] {
	if interval < 0 {
		return own
	}

	closed := false
	source := make(chan Item[T], channelCacheSize)
	go func() {
		var ok bool
		var item Item[T]
		ticker := time.NewTicker(interval)
		for {
			select {
			case item, ok = <-own.source:
				if !ok {
					closed = true
				}
			case <-ticker.C:
				source <- item
				ticker.Reset(interval)
			}

			if closed {
				break
			}
		}

		close(source)
	}()

	return &stream[T]{source: source}
}

// Clone 克隆当前数据流（异步）
// func (own *stream[T]) Clone() Stream[T] {
// 	source := make(chan Item[T], channelCacheSize)
// }

// Consume 一次性消费每个元素（异步）
func (own *stream[T]) Consume(consumeFunc func(item Item[T])) {
	if consumeFunc == nil {
		go own.Drain()
		return
	}

	go func() {
		for item := range own.source {
			consumeFunc(item)
		}
	}()
}

// Drain 排空数据（同步）
func (own *stream[T]) Drain(doBeforeFuncs ...func(item Item[T])) {
	if len(doBeforeFuncs) == 0 {
		for range own.source {
		}

		return
	}

	for item := range own.source {
		for _, do := range doBeforeFuncs {
			do(item)
		}
	}
}

// Receive 读取单个元素（同步）
func (own *stream[T]) Receive() (Item[T], bool) {
	for item := range own.source {
		return item, true
	}

	return nil, false
}

// Channel 获取数据流管道
func (own *stream[T]) Channel() <-chan Item[T] {
	return own.source
}
